/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package aiaudio.processing.reduce.klsh.kmeaning.hbase;

import aiaudio.processing.reduce.klsh.kmeaning.base.AlgorithmDataGrid;
import aiaudio.processing.reduce.klsh.kmeaning.base.Centroid;
import aiaudio.processing.reduce.klsh.kmeaning.base.Codepage;
import aiaudio.processing.reduce.klsh.kmeaning.base.Data;
import java.io.IOException;
import java.util.HashMap;
import java.util.ListIterator;
import org.apache.hadoop.hbase.client.Result;

/**
 *
 * @author nastya
 */
public abstract class AlgorithmDataGridOnDataSource extends AlgorithmDataGrid {

    private IRecordSource recordSource;

    public AlgorithmDataGridOnDataSource(IRecordSource recordSource) {
        this.recordSource = recordSource;
    }
    
    @Override
    public HashMap<Long, Data> testVector(Data vector) throws IOException {
        HashMap<Long, Data> data = new HashMap<Long, Data>();
        for (Codepage codepage : codepages) {
            Centroid minDistantionData = foundNearestCentroid(vector, codepage);
            data.put(codepage.getIndex(), new Data(minDistantionData.getKey()));
        }
        return data;
    }

    protected Centroid foundNearestCentroid(Data vector, Codepage codepage) throws IOException {
        
        Result testingVectorRecord = recordSource.getRecord(vector.getKey());
        ListIterator<Centroid> listIterator = codepage.listIterator();
        
        double minDistantion = Double.MAX_VALUE;
        Centroid minDistantionData = null;
        
        while(listIterator.hasNext()){
        
            Centroid centroid = listIterator.next();
            Result centroidRecord = recordSource.getRecord(centroid.getKey());
            double dist = calculateDistantion(testingVectorRecord, centroidRecord);
            
            if(dist < minDistantion){
                minDistantionData = centroid;
                minDistantion = dist;
            }
            
        }
        
        return minDistantionData;
    }

    protected abstract double calculateDistantion(Result testingVectorRecord, Result centroidRecord);
    
}
